package com.jch.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.Before;
import org.junit.Test;
import java.lang.Exception;
import java.util.Iterator;

public class DemoTest {

    Connection connection;
    HBaseAdmin admin;

    @Before
    public void init() throws Exception {

        //连接zookeeper
        Configuration configuration = HBaseConfiguration.create();

        configuration.set("hbase.zookeeper.quorum","hadoop101,hadoop102,hadoop103");
        connection = ConnectionFactory.createConnection(configuration);
        admin = (HBaseAdmin) connection.getAdmin();

    }

    //删除表
    @Test
    public void drop() throws Exception{
        //使表下线
        admin.disableTable(TableName.valueOf("user_info"));
        //删除表
        admin.deleteTable(TableName.valueOf("user_info"));
    }

    //创建表
    @Test
    public void create() throws Exception{

        TableDescriptorBuilder user_info = TableDescriptorBuilder.newBuilder(TableName.valueOf("user_info"));

        //列族
        ColumnFamilyDescriptor info = ColumnFamilyDescriptorBuilder.of(Bytes.toBytes("info"));
        ColumnFamilyDescriptor data = ColumnFamilyDescriptorBuilder.of(Bytes.toBytes("data"));

        //将表名和列族连接起来
        user_info.setColumnFamily(info);
        user_info.setColumnFamily(data);

        //构建表
        TableDescriptor build = user_info.build();
        admin.createTable(build);

    }


    @Test
    public void put() throws Exception{

        //添加数据
        Put put1 = new Put(Bytes.toBytes("rk001"));
        Put put2 = new Put(Bytes.toBytes("rk002"));
        Put put3 = new Put(Bytes.toBytes("rk003"));

        put1.addColumn(Bytes.toBytes("info"),Bytes.toBytes("name"),Bytes.toBytes("xiaoming"));
        put1.addColumn(Bytes.toBytes("data"),Bytes.toBytes("score"),Bytes.toBytes("90"));
        put2.addColumn(Bytes.toBytes("info"),Bytes.toBytes("name"),Bytes.toBytes("xiaohong"));
        put2.addColumn(Bytes.toBytes("data"),Bytes.toBytes("score"),Bytes.toBytes("85"));
        put3.addColumn(Bytes.toBytes("info"),Bytes.toBytes("name"),Bytes.toBytes("xiaohua"));
        put3.addColumn(Bytes.toBytes("data"),Bytes.toBytes("score"),Bytes.toBytes("88"));

        //获取表名
        Table user_info = connection.getTable(TableName.valueOf("user_info"));

        //添加数据
        user_info.put(put1);
        user_info.put(put2);
        user_info.put(put3);
    }

    //查询表中所有数据
    @Test
    public void show() throws Exception{

        //获取表名
        Table user_info = connection.getTable(TableName.valueOf("user_info"));
        Scan scan = new Scan();
        ResultScanner scanner = user_info.getScanner(scan);
        Iterator<Result> iterator = scanner.iterator();

        while (iterator.hasNext()){
            Result next = iterator.next();
            Cell[] cells = next.rawCells();

            for(Cell cell:cells){
                System.out.println("行键"+Bytes.toString(CellUtil.cloneRow(cell))+"\t"
                +"列族"+Bytes.toString(CellUtil.cloneFamily(cell))+"\t"
                +"列名"+Bytes.toString(CellUtil.cloneQualifier(cell))+"\t"
                +"值"+Bytes.toString(CellUtil.cloneValue(cell)));
            }

        }

    }

    //查询出user_info表中行键为rk002的数据
    @Test
    public void get() throws Exception{

        Scan scan = new Scan();
        Table user_info = connection.getTable(TableName.valueOf("user_info"));
        ResultScanner scanner = user_info.getScanner(scan);
        Iterator<Result> iterator = scanner.iterator();

        while (iterator.hasNext()){
            Result next = iterator.next();
            Cell[] cells = next.rawCells();

            for(Cell cell:cells){
                if (Bytes.toString(CellUtil.cloneRow(cell)).equals("rk001")){
                    System.out.println("行键"+Bytes.toString(CellUtil.cloneRow(cell))+"\t"
                    +"列族"+Bytes.toString(CellUtil.cloneFamily(cell))+"\t"
                    +"列名"+Bytes.toString(CellUtil.cloneQualifier(cell))+"\t"
                    +"值"+Bytes.toString(CellUtil.cloneValue(cell)));
                }
            }
        }
    }


    //删除
    @Test
    public void delete() throws Exception{

        //获取表名
        Table user_info = connection.getTable(TableName.valueOf("user_info"));
        Scan scan = new Scan();
        ResultScanner scanner = user_info.getScanner(scan);
        Iterator<Result> iterator = scanner.iterator();

        while (iterator.hasNext()){

            Result next = iterator.next();
            Cell[] cells = next.rawCells();

            for(Cell cell : cells){
                //删除表中行键为rk002的数据
                Delete delete = new Delete(Bytes.toBytes("rk002"));
                user_info.delete(delete);
            }
        }
    }

    //查看所有表
    @Test
    public void list() throws Exception{

        TableName[] tableNames = admin.listTableNames();

        for(TableName tableName:tableNames){
            //遍历表名
            System.out.println(tableName);
        }

    }

}
